package com.marktrace.consumer;

import com.marktrace.config.PulsarConfig;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @author Shellmine Dean Lance
 * @date 2022-08-29 10:46:55
 */
@Order(2)
@Component
public class PulsarConsumer implements CommandLineRunner {

    @Autowired
    private PulsarConfig pulsarConfig;

    @Override
    public void run(String... args) throws Exception {
        PulsarClient pulsarClient = PulsarClient.builder().serviceUrl(pulsarConfig.getClientUrl())//服务连接地址
                .authentication(AuthenticationFactory.token(pulsarConfig.getTenantToken()))
                .listenerName("external")
                .build();

        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .topic(pulsarConfig.getTenantTopic())
                .subscriptionName(pulsarConfig.getTenantId())
                .subscriptionType(SubscriptionType.Key_Shared)
                // 设置支持批量读取参数配置
                .batchReceivePolicy(
                        BatchReceivePolicy.builder()
                                .maxNumBytes(1024 * 1024)
                                .maxNumMessages(100)
                                .timeout(5, TimeUnit.SECONDS)
                                .build()
                ).subscribe();

        while (true) {
            //读取消息(批量)
            Messages<String> messages = consumer.batchReceive();
            //获取消息数据
            for (Message<String> message : messages) {
                String mqMsg = message.getValue();

                System.out.println("消息数据为:"+mqMsg);
                //例如：{"pushType":1,"data":{"event":2,"pushTime":"2022-08-25 17:25:10","warnDesc":{"markId":653595043893874688,"labelNumber":"100000100016592","farmId":653214654679482368,"farmName":"洮南王忠和养殖场","houseId":653329956331323392,"houseName":"1舍","columnId":653595040794284032,"columnName":"1号栏","createTime":"2022-05-06 02:43:45","deviceCode":"202108161010235","errMsg":null}}}
                //建议取到数据后，及时将数据转存至贵司内部的服务


                //3.3 ack确认
                consumer.acknowledge(message);
            }

        }

    }
}
